package mspbots.autotask.data.core.synclogs;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import mspbots.autotask.data.core.AbstractToolsUtil;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * mspbots.core.sync.autotask.badmessage.BadMessageReceiver
 *
 * @author <a href="https://github.com/vnobo">Alex bob</a>
 * @date Created by 2020/7/21
 */
@Log4j2
@Component
@RequiredArgsConstructor
public class BadMsgReceiver extends AbstractToolsUtil {

    private final BadMsgRepository messageRepository;

    @SuppressWarnings("unchecked")
    @RabbitListener(queues = "auto-task.bad.message")
    public void syncTask(Message<Object> message, @Headers Map<String, Object> headers) {

        log.error("AutoTask Sync bad message body {}", message.getPayload());
        log.error("AutoTask Sync bad message headers {}", headers);

        JsonNode msg = super.objectMapper.convertValue(message.getPayload(), JsonNode.class);
        long timestamp = Long.parseLong(headers.get("timestamp").toString());
        Instant instant = Instant.ofEpochMilli(timestamp);

        List<Map<String, ?>> xDeath = (List<Map<String, ?>>) headers.get("x-death");
        Map<String, ?> xDeathMap = xDeath.get(0);
        Long count = (Long) xDeathMap.get("count");
        Date time = (Date) xDeathMap.get("time");
        List<String> keys = (List<String>) xDeathMap.get("routing-keys");

        BadMsg badMessage = BadMsg.builder()
                .tenantId(msg.get("tenantId").longValue())
                .msgId(headers.get("id").toString())
                .msg(msg)
                .count(count.intValue())
                .exchange(xDeathMap.get("exchange").toString())
                .queue(xDeathMap.get("queue").toString())
                .reason(xDeathMap.get("reason").toString())
                .routingKeys(StringUtils.collectionToCommaDelimitedString(keys))
                .errTime(LocalDateTime.ofInstant(time.toInstant(), ZoneId.systemDefault()))
                .msgTime(LocalDateTime.ofInstant(instant, ZoneId.systemDefault()))
                .build();

        this.messageRepository.save(badMessage).block(Duration.ofSeconds(10));
    }

}
